package com.wormpex.data.root.common;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.net.URL;
import java.text.SimpleDateFormat;
import java.util.*;

public class KafkaProducerUtil{
    private static Logger logger = LoggerFactory.getLogger(KafkaProducerUtil.class);
    private static final String KAFKA_CONFIG_FILE_NAME = "kafka.properties";

    public static KafkaProducer getKafkaProducer() {
        KafkaProducer<String, String> kafkaProducer ;

        try {
            logger.info("Try to create kafka producer");
            Map<String, Integer> topicCountMap = new HashMap<>();

            kafkaProducer = new KafkaProducer<String, String>(generateKafkaConfig());

            logger.info("Successfully create kafka producer");
            return kafkaProducer;
        } catch (IOException exp) {
            logger.error("Failed to init kafka producer", exp);
            throw new RuntimeException(exp);
        } catch (RuntimeException exp) {
            logger.error("Got RuntimeException", exp);
            throw new RuntimeException(exp);
        }
    }

    private static Properties generateKafkaConfig() throws IOException {
        logger.info("Generate Kafka producer Config");
        URL fileUrl = KafkaProducer.class.getClassLoader().getResource(KAFKA_CONFIG_FILE_NAME);
        if (fileUrl != null) {
            InputStream in = fileUrl.openStream();
            Properties configProperties = new Properties();
            configProperties.load(in);
            in.close();
            //configProperties.setProperty("producer.id", producerId);
            return configProperties;

        } else {
            logger.error("Can not find {}, fail to initialize kafka producer", KAFKA_CONFIG_FILE_NAME);
            throw new RuntimeException("Failed to load kafka configuration file");
        }
    }

    private static String generateProducerId(String topicName, int index) {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MMdd-HHmmss");
        Calendar now = Calendar.getInstance();
        String timestampString = format.format(now.getTime());
        String machineGUID = ManagementFactory.getRuntimeMXBean().getName() + "_" + topicName + Thread.currentThread().getId() + "_" + index;

        return String.format("%s@%s", machineGUID, timestampString);
    }
}